package twitterhadoop.hadoop.io;

import java.io.IOException;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Properties;
import java.util.Random;
import java.util.Set;
import java.util.regex.Pattern;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import twitter4j.Query;
import twitter4j.QueryResult;
import twitter4j.Status;
import twitter4j.Twitter;
import twitter4j.TwitterException;
import twitter4j.TwitterFactory;
import twitter4j.auth.AccessToken;
import twitter4j.conf.ConfigurationBuilder;

public class TwitterRecordReader extends RecordReader<LongWritable, TextArrayWritable> {
	private static final Log LOG = LogFactory.getLog(TwitterRecordReader.class);
	private static Random randomGen = new Random(System.currentTimeMillis());
	
	private static final Pattern nonASCII = Pattern.compile("[^\\p{Alnum}]");
	
	private Twitter twitter;
	private TwitterInputSplit split;
	private TaskAttemptContext context;
	private QueryResult queryResult;
	private Iterator<Status> tweetsBufferIterator;
	private boolean finishedAllTweets;
	private int tweetsCounter = 0;
	
	private LongWritable currKey;
	private TextArrayWritable currVal;
	
	private Twitter createTwitterFromRandomCredentials() {
		Twitter t = null;
				
		Properties prop = loadRandomProperties();
		
		if (prop != null) {
			ConfigurationBuilder cb = new ConfigurationBuilder();
			cb.setDebugEnabled(Boolean.parseBoolean(prop.getProperty("debug")))
				.setOAuthConsumerKey(prop.getProperty("oauth.consumerKey"))
				.setOAuthConsumerSecret(prop.getProperty("oauth.consumerSecret"))
				.setOAuthAccessToken(prop.getProperty("oauth.accessToken"))
				.setOAuthAccessTokenSecret(prop.getProperty("oauth.accessTokenSecret"));
			
			TwitterFactory tf = new TwitterFactory(cb.build());
			t = tf.getInstance();
			
		} else {
			t = new TwitterFactory().getInstance();
		}

		return t;
	}
	
	private Properties loadRandomProperties() {
		Properties prop = null;
		int fileIndex = randomGen.nextInt(9) + 1;
		String twitterConfigFilename = String.format("twitter4j.%d.properties", fileIndex);
		
		try {
			prop = new Properties();
			// File resourceFolder = new File("res/");
			// File[] twitterConfigFiles = resourceFolder.listFiles();
			// int fileIndex = randomGen.nextInt(twitterConfigFiles.length);
			
			//File configFile = twitterConfigFiles[fileIndex];
			
			LOG.info("reading twitter4j file: " + twitterConfigFilename);
			
			//InputStream in = new FileInputStream(configFile);
			
			prop.load(TwitterRecordReader.class.getClassLoader().getResourceAsStream(twitterConfigFilename));
			//prop.load(in);
			
		} catch (IOException e) {
			//LOG.error("Cannot read twitter configuration file: " + configFile.getName());
			LOG.error("Cannot read twitter configuration file: " + twitterConfigFilename);
			e.printStackTrace();
		}
		
		
		return prop;
	}
	
	@Override
	public void initialize(InputSplit inSplit, TaskAttemptContext context) throws IOException, InterruptedException {
		this.split = (TwitterInputSplit) inSplit;
		this.context = context;
        try {
        	LOG.info(String.format("Initializing: q=%s, since=%s, until=%s", split.getQuery(), split.getStartDate(), split.getEndDate()));
            // gets Twitter instance with default credentials
            this.twitter = createTwitterFromRandomCredentials();
            //User user = twitter.verifyCredentials();
            Query q = new Query(split.getQuery());
            q.setSince(split.getStartDate());
            q.setUntil(split.getEndDate());
            queryResult = twitter.search(q);
            LOG.info(String.format("%d tweets received", queryResult.getTweets().size()));
            tweetsBufferIterator = queryResult.getTweets().iterator();
            
            currKey = new LongWritable();
            currVal = new TextArrayWritable();
            finishedAllTweets = false;
            
            
        } catch (TwitterException te) {
            te.printStackTrace();
            System.out.println("Failed to get timeline: " + te.getMessage());
            System.exit(-1);
        }	
		
	}

	@Override
	public boolean nextKeyValue() throws IOException, InterruptedException {
		if (!tweetsBufferIterator.hasNext())  {
			if (queryResult.hasNext()) {
				try {
					LOG.info("twitter-cache: fetching another buffer..." + queryResult.getQuery());
					
		            queryResult = twitter.search(queryResult.nextQuery());
		            tweetsBufferIterator = queryResult.getTweets().iterator();
		        } catch (TwitterException te) {
		            te.printStackTrace();
		            System.out.println("Failed to get timeline: " + te.getMessage());
		            System.exit(-1);
		        }	
			}
		}
		 
		if (this.tweetsCounter >= this.split.getLimit()) {
			LOG.info(String.format("Tweets limit reached: %d tweets read.", this.tweetsCounter));
		} else if (tweetsBufferIterator.hasNext()) {
			Status st = tweetsBufferIterator.next();
			this.currKey.set(st.getId());
			
			String[] tweetStr = st.getText().toLowerCase().split(" ");
			for (int i = 0; i < tweetStr.length; i++) {
				String temp = nonASCII.matcher(tweetStr[i]).replaceAll("");
				tweetStr[i] = temp;
			}
			
			tweetStr = removeDuplicatedAndTrunc(tweetStr);
			
			Arrays.sort(tweetStr);
			Text[] tweetText = new Text[tweetStr.length];
			for (int i = 0; i < tweetText.length; i++) {
				if (tweetStr[i].length() > 1) {
					tweetText[i] = new Text(tweetStr[i].replaceAll("[^a-zA-Z0-9]", ""));
				}
				
			}
			
			this.currVal.set(tweetText);
			this.currVal.setArray(tweetStr);
			LOG.debug("tweet => " + st.getText().toLowerCase());
			this.tweetsCounter++;
			return true;
		}
		
        finishedAllTweets = true;
		return false;
	}

	/**
	 * @param tweetStr
	 * @return
	 */
	private String[] removeDuplicatedAndTrunc(String[] tweetStr) {
		// remove duplicates 
		Set<String> tempSet = new HashSet<String>();
		for (String word : tweetStr) {
			if (word.length() > 1) {
				tempSet.add(word);				
			}
		}
		return tempSet.toArray(new String[0]);
	}

	@Override
	public LongWritable getCurrentKey() throws IOException, InterruptedException {
		return this.currKey;
	}

	@Override
	public TextArrayWritable getCurrentValue() throws IOException, InterruptedException {
		return this.currVal;
	}

	@Override
	public float getProgress() throws IOException, InterruptedException {
		if (finishedAllTweets)
			return 1.0f;
		
		return 0.0f;
	}

	@Override
	public void close() throws IOException {
		// Do nothing
	}

}
